package com.simon.cdc.converters;

import io.debezium.spi.converter.CustomConverter;
import io.debezium.spi.converter.RelationalColumn;
import org.apache.kafka.connect.data.SchemaBuilder;

import java.time.*;
import java.time.format.DateTimeFormatter;
import java.util.Properties;

public class DateTimeFormatterConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {

    private DateTimeFormatter datetimeFormatter;
    private boolean trimMillis;

    /**
     * 根据提供的属性配置日期时间格式化器
     *
     * @param props 包含配置属性的Properties对象，用于定制日期时间格式化器的行为
     */
    @Override
    public void configure(Properties props) {
        // 从属性中获取时区设置，如果没有设置，则默认为"UTC"
        String timezone = props.getProperty("timezone", "UTC");
        // 从属性中获取是否修剪毫秒的设置，如果没有设置，则默认为"true"
        this.trimMillis = Boolean.parseBoolean(props.getProperty("trim.millis", "true"));

        // 初始化格式化器，使用提供的日期时间格式和时区设置
        this.datetimeFormatter = DateTimeFormatter.ofPattern(props.getProperty("datetime.format", "yyyy-MM-dd HH:mm:ss"))
                .withZone(ZoneId.of(timezone));
    }

    /**
     * 根据RelationalColumn对象的类型，注册相应的转换器
     * 此方法旨在处理时间相关的数据类型，将它们转换为适当的字符串表示形式
     *
     * @param column       表示数据库列的RelationalColumn对象，用于获取列的类型信息
     * @param registration 用于注册转换器的ConverterRegistration对象，通过它来绑定转换逻辑
     */
    @Override
    public void converterFor(RelationalColumn column, ConverterRegistration<SchemaBuilder> registration) {

        // 将列的类型名称转换为大写，以便后续比较
        String typeName = column.typeName().toUpperCase();

        // 判断列类型是否为TIMESTAMP或DATETIME
        if (typeName.contains("TIMESTAMP") || typeName.contains("DATETIME")) {
            // 注册一个字符串类型的转换器，根据值的类型进行不同的处理
            registration.register(SchemaBuilder.string(), value -> {
                if (value == null) return null;

                // 如果值是Long类型，视为时间戳
                if (value instanceof Long) {
                    // 处理时间戳
                    return datetimeFormatter.format(Instant.ofEpochMilli((Long) value));
                } else if (value instanceof String strValue) {
                    // 如果值是字符串类型，检查是否需要移除最后的".0"
                    return trimMillis && strValue.endsWith(".0") ?
                            strValue.substring(0, strValue.length() - 2) : strValue;
                }
                // 其他情况直接转换为字符串
                return value.toString();
            });
        }
        // 判断列类型是否为DATE
        else if (typeName.contains("DATE")) {
            // 注册一个字符串类型的转换器，直接将值转换为字符串
            // DATE类型通常不会有.0问题
            registration.register(SchemaBuilder.string(), value -> {
                if (value == null) return null;
                return value.toString();
            });
        }
        // 判断列类型是否为TIME
        else if (typeName.contains("TIME")) {
            // 注册一个字符串类型的转换器，直接将值转换为字符串
            // TIME类型通常不会有.0问题
            registration.register(SchemaBuilder.string(), value -> {
                if (value == null) return null;
                return value.toString();
            });
        }
    }
}